package concurrent.c_012_ThreadPool;

import util.ThreadUtil;
import util.TimeUtil;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import static java.util.stream.Collectors.toList;

/**
 * https://www.bilibili.com/video/BV1np4y1C7Yf?p=195
 * https://blog.csdn.net/f641385712/article/details/83580886
 * https://blog.csdn.net/f641385712/article/details/83618189
 * JDK1.8 才新加入的一个实现类，实现了 Future, CompletionStage 2个接口（CompletionStage 接口也是 1.8 才提供的）
 * <p>
 * CompletableFuture 的简单介绍:：
 * 当一个 Future 可能需要显示地完成时，使用 CompletionStage 接口去支持完成时触发的函数和操作。
 * 当 2 个以上线程同时尝试完成、异常完成、取消一个 CompletableFuture 时，只有一个能成功。
 * <p>
 * CompletableFuture 实现了 CompletionStage 接口的如下策略：
 * * 为了完成当前的 CompletableFuture 接口或者其他完成方法的回调函数的线程，提供了非异步的完成操作
 * * 没有显式入参，Executor 的所有 async 方法都使用 ForkJoinPool.commonPool()
 * * 为了简化监视、调试和跟踪，所有生成的异步任务都是标记接口 AsynchronousCompletionTask 的实例
 * * 所有的 CompletionStage 方法都是独立于其它共有方法实现的，因此一个方法的行为不会受到子类中其它方法的覆盖
 * <p>
 * CompletableFuture 实现了 Futurre 接口的如下策略：
 * * CompletableFuture 无法直接控制完成，所以 cancel 操作被视为是另一种异常完成形式。
 * * 方法 isCompletedExceptionally 可以用来确定一个 CompletableFuture 是否以任何异常的方式完成。
 * * 以一个 CompletionException 为例，方法 get() 和 get(long,TimeUnit) 抛出一个 ExecutionException，对应 CompletionException。
 * * 为了在大多数上下文中简化用法，这个类还定义了方法 join() 和 getNow，而不是直接在这些情况中直接抛出 CompletionException。
 * <p>
 * CompletionStage 接口实现流式编程：JDK8 新增接口，此接口包含 38 个方法。这些方法主要是为了支持函数式编程中流式处理。
 * * 其中 supplyAsync 用于有返回值的任务，runAsync 则用于没有返回值的任务。
 * * Executor 参数可以手动指定线程池，否则默认 ForkJoinPool.commonPool() 系统级公共线程池(这些线程都是Daemon线程，
 * 主线程结束 Daemon 线程不结束，只有 JVM 关闭时，生命周期终止)
 * <p>
 * JDK8 的 CompletableFuture 自带多任务组合方法 allOf 和 anyOf
 * * allOf 是等待所有任务完成，构造后 CompletableFuture 完成
 * * anyOf 是只要有一个任务完成，构造后 CompletableFuture 就完成
 */
public class T09_CompletableFuture {
    // 定长 10 线程池
    static ExecutorService exs = Executors.newFixedThreadPool(10);

    public static void main(String[] args) {
        CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(() -> {
            ThreadUtil.sleepMillis(3000);
            return 100;
        });
        CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(() -> {
            ThreadUtil.sleepMillis(3000);
            return 200;
        });
        CompletableFuture<List<Integer>> resultList = ThreadUtil.sequence(Arrays.asList(future1, future2));
        System.out.println(resultList.join()); // [100, 200]

        demo();
        // demo1();
        // demo2();
    }

    public static void demo() {
        try {
            CompletableFuture<String> completableFuture = new CompletableFuture();
            new Thread(() -> { // 自己开个线程去执行 执行完把结果告诉 completableFuture 即可
                try {
                    Thread.sleep(3000); // 模拟执行耗时任务
                    // System.out.println(1 / 0);
                } catch (Exception e) {
                    completableFuture.completeExceptionally(e); // 告诉 completableFuture 任务发生异常了
                }
                completableFuture.complete("ok"); // 任务已经完成，把结果告诉 completableFuture
            }).start();
            System.out.println(completableFuture.get()); // 获取任务结果，如果没有完成会一直阻塞等待


            CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> "x", exs)
                    .handle((integer, throwable) -> Objects.requireNonNullElse(integer, "处理结果返回"));
            System.out.println(future.get());

            CompletableFuture<String> chain = CompletableFuture.supplyAsync(() -> "x", exs)
                    .thenApply(s -> s + "y")
                    .thenApplyAsync(s -> s + "z", exs);
            System.out.println("链式处理：" + chain.get());


            CompletableFuture.supplyAsync(() -> "x", exs).thenRunAsync(() -> System.out.println("顺序编排"), exs);
            CompletableFuture.supplyAsync(() -> "x", exs).thenAcceptAsync(s -> System.out.println("获取返回值并顺序编排：" + s), exs);


            CompletableFuture<String> taskOne = CompletableFuture.supplyAsync(() -> "task one", exs);
            CompletableFuture<String> taskTwo = CompletableFuture.supplyAsync(() -> "task two", exs);
            // 两个都完成
            taskOne.runAfterBothAsync(taskTwo, () -> System.out.println("taskOne 和 taskTwo 都已完成"), exs); // 不感知结果
            taskOne.thenAcceptBothAsync(taskTwo, (one, two) -> System.out.println("结果：" + one + " " + two), exs); // 感知结果
            taskOne.thenCombineAsync(taskTwo, (one, two) -> one + " 和 " + two + " 已完成", exs)
                    .whenCompleteAsync((v, e) -> System.out.println(v)); // 感知结果并有返回值
            // 只要一个完成
            taskOne.runAfterEitherAsync(taskTwo, () -> System.out.println("taskOne 或 taskTwo 已完成"), exs); // 不感知结果
            taskOne.acceptEitherAsync(taskTwo, s -> System.out.println("结果：" + s), exs); // 感知结果
            taskOne.applyToEitherAsync(taskTwo, s -> s + " 已完成", exs)
                    .whenCompleteAsync((v, e) -> System.out.println(v)); // 感知结果并有返回值
            // 任意编排
            CompletableFuture<Void> allOf = CompletableFuture.allOf(taskOne, taskTwo); // 所有完成
            // allOf.get();
            allOf.join();
            System.out.println("main 等待 taskOne 和 taskTwo 完成：" + taskOne.get() + taskTwo.get());
            CompletableFuture<Object> anyOf = CompletableFuture.anyOf(taskOne, taskTwo); // 任意一个完成
            System.out.println("taskOne 或 taskTwo 完成：" + anyOf.get());
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            // exs.shutdown();
            System.out.println(exs);
        }
    }

    /**
     * 循环创建 CompletableFuture list,调用 sequence() 组装返回一个有返回值的 CompletableFuture，返回结果 get() 获取
     */
    public static void demo1() {
        long start = System.currentTimeMillis();
        // 结果集
        List<String> list;
        List<String> list2 = new ArrayList<>();
        List<CompletableFuture<String>> futureList = new ArrayList<>();
        List<Integer> taskList = Arrays.asList(2, 1, 3, 4, 5, 6, 7, 8, 9, 10);
        try {
            // 方式一：循环创建 CompletableFuture list, 然后组装，组装返回一个有返回值的 CompletableFuture，返回结果 get() 获取
            for (int i = 0; i < taskList.size(); i++) {
                final int j = i;
                // 异步执行，拿到每个有返回值的 CompletableFuture 对象
                CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> calc(taskList.get(j)), exs)
                        // Integer 转换字符串，thenAccept 只接受不返回，不影响结果
                        .thenApply(e -> Integer.toString(e))
                        // 如需获取任务完成先后顺序，此处代码即可
                        // .whenComplete((v, e) -> { // 使用当前线程运行，只能感知异常
                        .whenCompleteAsync((v, e) -> { // 使用线程池线程运行，只能感知异常
                            System.out.println("任务 " + v + " 完成！result = " + v + "，异常 e = " + e + "，" + TimeUtil.nowDateTime());
                            list2.add(v);
                        }, exs)
                        .exceptionally(throwable -> "异常默认返回"); // 能感知异常并处理返回结果
                futureList.add(future);
            }

            // 流式获取结果：此处是根据任务添加顺序获取的结果
            // 1.构造一个空 CompletableFuture，子任务数为入参任务 list size
            CompletableFuture<Void> allDoneFuture = CompletableFuture.allOf(futureList.stream()
                    .filter(Objects::nonNull).collect(toList()).toArray(new CompletableFuture[futureList.size()]));
            // 2.流式（总任务完成后，每个子任务 join 取结果，后转换为 list）
            list = allDoneFuture.thenApply(v -> futureList.stream().map(CompletableFuture::join).collect(toList())).get();
            // 流式获取结果：此处是根据任务添加顺序获取的结果

            System.out.println("任务完成先后顺序，结果 list2 = " + list2 + "；任务提交顺序，结果 list = " + list + "，耗时 = " + (System.currentTimeMillis() - start));
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            exs.shutdown();
        }
    }

    /**
     * * 推荐
     * 全流式处理转换成 CompletableFuture[] + allOf 组装成一个无返回值 CompletableFuture，join 等待执行完毕。返回结果 whenComplete 获取
     */
    public static void demo2() {
        long start = System.currentTimeMillis();
        // 结果集
        List<String> list = new ArrayList<>();
        List<String> list2 = new ArrayList<>();
        final List<Integer> taskList = Arrays.asList(2, 1, 3, 4, 5, 6, 7, 8, 9, 10);
        try {
            // 方式二：全流式处理转换成 CompletableFuture[] + 组装成一个无返回值 CompletableFuture，join 等待执行完毕。返回结果 whenComplete 获取
            CompletableFuture<Integer>[] cfs = taskList.stream().map(i ->
                    // 把计算任务交给 CompletableFuture 异步去处理执行
                    CompletableFuture.supplyAsync(() -> calc(i), exs)
                            // 把计算完成结果做 Function 处理：此处是转换成了字符串
                            .thenApply(h -> Integer.toString(h))
                            // 如需获取任务完成先后顺序，此处代码即可，会先处理先完成的任务，后处理后完成的任务，使用起来比 CompletionService 确实方便不少
                            .whenComplete((v, e) -> {
                                System.out.println("任务 " + v + " 完成！result = " + v + "，异常 e = " + e + "，" + TimeUtil.nowDateTime());
                                list2.add(v);
                            })).toArray(CompletableFuture[]::new); // 此处直接 toArray，不 toList 了

            // 等待总任务完成，但是封装后无返回值，必须自己 whenComplete() 获取，此处使用 join 来获取结果
            CompletableFuture.allOf(cfs).join();
            System.out.println("任务完成先后顺序，结果 list2 = " + list2 + "；任务提交顺序，结果 list = " + list + "，耗时 = " + (System.currentTimeMillis() - start));
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            exs.shutdown();
        }
    }

    public static Integer calc(Integer i) { // 模拟任务的耗时方法
        try {
            if (i == 1) { // 任务 1 耗时 3 秒
                Thread.sleep(3000);
            } else if (i == 5) { // 任务 5 耗时 5 秒
                Thread.sleep(5000);
            } else { // 其它任务耗时 1 秒
                Thread.sleep(1000);
            }
            System.out.println("task 线程：" + Thread.currentThread().getName() + "任务 i = " + i + "，完成！+" + TimeUtil.nowDateTime());
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return i;
    }
}
